//! # Kubernetes 服务发现
//!
//! 基于 Kubernetes 服务和端点的自动服务发现功能。

use std::collections::{HashMap, BTreeMap};
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use futures::stream::StreamExt;
use kube::{
    Api, Client,
    runtime::{watcher, WatchStreamExt},
};
use k8s_openapi::api::core::v1::{Service, Endpoints, Pod};
use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;
use tracing::{info, debug, warn, error};

use crate::client::KubernetesClient;
use crate::error::{K8sError, K8sResult};

/// 服务信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ServiceInfo {
    /// 服务名称
    pub name: String,
    /// 命名空间
    pub namespace: String,
    /// 服务类型
    pub service_type: String,
    /// 集群 IP
    pub cluster_ip: Option<String>,
    /// 外部 IP
    pub external_ips: Vec<String>,
    /// 端口信息
    pub ports: Vec<ServicePort>,
    /// 标签
    pub labels: HashMap<String, String>,
    /// 注解
    pub annotations: HashMap<String, String>,
    /// 端点列表
    pub endpoints: Vec<ServiceEndpoint>,
}

/// 服务端口信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ServicePort {
    /// 端口名称
    pub name: Option<String>,
    /// 端口号
    pub port: i32,
    /// 目标端口
    pub target_port: Option<i32>,
    /// 协议
    pub protocol: String,
    /// 节点端口（NodePort 服务）
    pub node_port: Option<i32>,
}

/// 服务端点信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ServiceEndpoint {
    /// IP 地址
    pub ip: String,
    /// 端口
    pub port: i32,
    /// 是否就绪
    pub ready: bool,
    /// 节点名称
    pub node_name: Option<String>,
    /// Pod 名称
    pub pod_name: Option<String>,
}

/// 服务发现接口
#[async_trait]
pub trait ServiceDiscovery: Send + Sync {
    /// 发现服务
    async fn discover_services(&self, namespace: &str) -> K8sResult<Vec<ServiceInfo>>;
    
    /// 获取服务详情
    async fn get_service(&self, namespace: &str, name: &str) -> K8sResult<ServiceInfo>;
    
    /// 监听服务变化
    async fn watch_services<F>(&self, namespace: &str, callback: F) -> K8sResult<()>
    where
        F: Fn(ServiceEvent) + Send + Sync + 'static;
}

/// 服务事件
#[derive(Debug, Clone)]
pub enum ServiceEvent {
    /// 服务添加
    Added(ServiceInfo),
    /// 服务修改
    Modified(ServiceInfo),
    /// 服务删除
    Deleted(ServiceInfo),
}

/// Kubernetes 服务发现实现
pub struct K8sServiceDiscovery {
    /// Kubernetes 客户端
    client: Arc<KubernetesClient>,
    /// 服务缓存
    service_cache: Arc<RwLock<HashMap<String, ServiceInfo>>>,
}

impl K8sServiceDiscovery {
    /// 创建新的服务发现实例
    pub fn new(client: Arc<KubernetesClient>) -> Self {
        Self {
            client,
            service_cache: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// 从 Kubernetes Service 资源创建 ServiceInfo
    async fn service_to_info(&self, service: &Service, namespace: &str) -> K8sResult<ServiceInfo> {
        let metadata = &service.metadata;
        
        let spec = service.spec.as_ref()
            .ok_or_else(|| K8sError::service_discovery_error("Service 缺少 spec"))?;

        let name = metadata.name.as_ref()
            .ok_or_else(|| K8sError::service_discovery_error("Service 缺少 name"))?
            .clone();

        // 获取端口信息
        let ports = spec.ports.as_ref().unwrap_or(&vec![])
            .iter()
            .map(|port| ServicePort {
                name: port.name.clone(),
                port: port.port,
                target_port: port.target_port.as_ref().and_then(|tp| {
                    match tp {
                        k8s_openapi::apimachinery::pkg::util::intstr::IntOrString::Int(i) => Some(*i),
                        k8s_openapi::apimachinery::pkg::util::intstr::IntOrString::String(_) => None,
                    }
                }),
                protocol: port.protocol.as_ref().unwrap_or(&"TCP".to_string()).clone(),
                node_port: port.node_port,
            })
            .collect();

        // 获取端点信息
        let endpoints = self.get_service_endpoints(namespace, &name).await
            .unwrap_or_else(|e| {
                warn!("无法获取服务 {}/{} 的端点: {}", namespace, name, e);
                vec![]
            });

        Ok(ServiceInfo {
            name,
            namespace: namespace.to_string(),
            service_type: spec.type_.as_ref().unwrap_or(&"ClusterIP".to_string()).clone(),
            cluster_ip: spec.cluster_ip.clone(),
            external_ips: spec.external_ips.as_ref().unwrap_or(&vec![]).clone(),
            ports,
            labels: metadata.labels.as_ref().unwrap_or(&BTreeMap::new()).iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
            annotations: metadata.annotations.as_ref().unwrap_or(&BTreeMap::new()).iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
            endpoints,
        })
    }

    /// 获取服务端点
    async fn get_service_endpoints(&self, namespace: &str, service_name: &str) -> K8sResult<Vec<ServiceEndpoint>> {
        let endpoints_api: Api<Endpoints> = Api::namespaced(self.client.client().clone(), namespace);
        
        let endpoints = endpoints_api.get(service_name).await
            .map_err(|e| K8sError::service_discovery_error(format!("无法获取端点: {}", e)))?;

        let mut service_endpoints = Vec::new();

        if let Some(subsets) = endpoints.subsets {
            for subset in subsets {
                let ports: Vec<i32> = subset.ports
                    .unwrap_or_default()
                    .iter()
                    .map(|p| p.port)
                    .collect();

                // 就绪的端点
                if let Some(addresses) = subset.addresses {
                    for address in addresses {
                        for &port in &ports {
                            service_endpoints.push(ServiceEndpoint {
                                ip: address.ip.clone(),
                                port,
                                ready: true,
                                node_name: address.node_name.clone(),
                                pod_name: address.target_ref.as_ref()
                                    .and_then(|tr| tr.name.clone()),
                            });
                        }
                    }
                }

                // 未就绪的端点
                if let Some(not_ready_addresses) = subset.not_ready_addresses {
                    for address in not_ready_addresses {
                        for &port in &ports {
                            service_endpoints.push(ServiceEndpoint {
                                ip: address.ip.clone(),
                                port,
                                ready: false,
                                node_name: address.node_name.clone(),
                                pod_name: address.target_ref.as_ref()
                                    .and_then(|tr| tr.name.clone()),
                            });
                        }
                    }
                }
            }
        }

        Ok(service_endpoints)
    }

    /// 过滤服务
    pub fn filter_services(&self, services: Vec<ServiceInfo>, filter: ServiceFilter) -> Vec<ServiceInfo> {
        services.into_iter()
            .filter(|service| filter.matches(service))
            .collect()
    }

    /// 获取健康的端点
    pub fn get_healthy_endpoints(service: &ServiceInfo) -> Vec<&ServiceEndpoint> {
        service.endpoints.iter()
            .filter(|endpoint| endpoint.ready)
            .collect()
    }
}

#[async_trait]
impl ServiceDiscovery for K8sServiceDiscovery {
    async fn discover_services(&self, namespace: &str) -> K8sResult<Vec<ServiceInfo>> {
        debug!("发现命名空间 {} 中的服务", namespace);

        let services_api: Api<Service> = Api::namespaced(self.client.client().clone(), namespace);
        let services = services_api.list(&Default::default()).await
            .map_err(|e| K8sError::service_discovery_error(format!("无法列出服务: {}", e)))?;

        let mut service_infos = Vec::new();

        for service in services.items {
            match self.service_to_info(&service, namespace).await {
                Ok(service_info) => {
                    service_infos.push(service_info);
                }
                Err(e) => {
                    warn!("转换服务信息失败: {}", e);
                }
            }
        }

        info!("在命名空间 {} 中发现 {} 个服务", namespace, service_infos.len());
        Ok(service_infos)
    }

    async fn get_service(&self, namespace: &str, name: &str) -> K8sResult<ServiceInfo> {
        debug!("获取服务 {}/{}", namespace, name);

        let services_api: Api<Service> = Api::namespaced(self.client.client().clone(), namespace);
        let service = services_api.get(name).await
            .map_err(|e| K8sError::resource_not_found("Service", name, namespace))?;

        self.service_to_info(&service, namespace).await
    }

    async fn watch_services<F>(&self, namespace: &str, callback: F) -> K8sResult<()>
    where
        F: Fn(ServiceEvent) + Send + Sync + 'static
    {
        info!("开始监听命名空间 {} 中的服务变化", namespace);

        let services_api: Api<Service> = Api::namespaced(self.client.client().clone(), namespace);
        let watcher = watcher(services_api, Default::default());

        let callback = Arc::new(callback);
        
        tokio::spawn({
            let client = self.client.clone();
            let namespace = namespace.to_string();
            
            async move {
                let mut stream = watcher.boxed();
                
                while let Some(event) = stream.next().await {
                    match event {
                        Ok(watcher::Event::Applied(service)) => {
                            if let Ok(service_info) = Self::create_service_info_from_k8s(&client, &service, &namespace).await {
                                callback(ServiceEvent::Added(service_info));
                            }
                        }
                        Ok(watcher::Event::Deleted(service)) => {
                            if let Ok(service_info) = Self::create_service_info_from_k8s(&client, &service, &namespace).await {
                                callback(ServiceEvent::Deleted(service_info));
                            }
                        }
                        Err(e) => {
                            error!("服务监听错误: {}", e);
                        }
                        _ => {}
                    }
                }
            }
        });

        Ok(())
    }
}

impl K8sServiceDiscovery {
    async fn create_service_info_from_k8s(
        client: &Arc<KubernetesClient>,
        service: &Service,
        namespace: &str,
    ) -> K8sResult<ServiceInfo> {
        let discovery = K8sServiceDiscovery::new(client.clone());
        discovery.service_to_info(service, namespace).await
    }
}

/// 服务过滤器
pub struct ServiceFilter {
    /// 标签选择器
    pub label_selector: Option<HashMap<String, String>>,
    /// 服务类型过滤
    pub service_types: Option<Vec<String>>,
    /// 名称匹配
    pub name_pattern: Option<String>,
}

impl ServiceFilter {
    /// 创建新的过滤器
    pub fn new() -> Self {
        Self {
            label_selector: None,
            service_types: None,
            name_pattern: None,
        }
    }

    /// 添加标签选择器
    pub fn with_labels(mut self, labels: HashMap<String, String>) -> Self {
        self.label_selector = Some(labels);
        self
    }

    /// 添加服务类型过滤
    pub fn with_service_types(mut self, types: Vec<String>) -> Self {
        self.service_types = Some(types);
        self
    }

    /// 添加名称模式
    pub fn with_name_pattern(mut self, pattern: String) -> Self {
        self.name_pattern = Some(pattern);
        self
    }

    /// 检查服务是否匹配过滤条件
    pub fn matches(&self, service: &ServiceInfo) -> bool {
        // 检查标签选择器
        if let Some(ref selector) = self.label_selector {
            for (key, value) in selector {
                if service.labels.get(key) != Some(value) {
                    return false;
                }
            }
        }

        // 检查服务类型
        if let Some(ref types) = self.service_types {
            if !types.contains(&service.service_type) {
                return false;
            }
        }

        // 检查名称模式
        if let Some(ref pattern) = self.name_pattern {
            if !service.name.contains(pattern) {
                return false;
            }
        }

        true
    }
}

impl Default for ServiceFilter {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_service_filter() {
        let service = ServiceInfo {
            name: "my-service".to_string(),
            namespace: "default".to_string(),
            service_type: "ClusterIP".to_string(),
            cluster_ip: Some("10.0.0.1".to_string()),
            external_ips: vec![],
            ports: vec![],
            labels: {
                let mut labels = HashMap::new();
                labels.insert("app".to_string(), "web".to_string());
                labels
            },
            annotations: HashMap::new(),
            endpoints: vec![],
        };

        // 测试标签过滤
        let filter = ServiceFilter::new()
            .with_labels({
                let mut labels = HashMap::new();
                labels.insert("app".to_string(), "web".to_string());
                labels
            });
        assert!(filter.matches(&service));

        // 测试服务类型过滤
        let filter = ServiceFilter::new()
            .with_service_types(vec!["ClusterIP".to_string()]);
        assert!(filter.matches(&service));

        // 测试名称模式过滤
        let filter = ServiceFilter::new()
            .with_name_pattern("service".to_string());
        assert!(filter.matches(&service));
    }

    #[test]
    fn test_healthy_endpoints() {
        let service = ServiceInfo {
            name: "test".to_string(),
            namespace: "default".to_string(),
            service_type: "ClusterIP".to_string(),
            cluster_ip: None,
            external_ips: vec![],
            ports: vec![],
            labels: HashMap::new(),
            annotations: HashMap::new(),
            endpoints: vec![
                ServiceEndpoint {
                    ip: "10.0.0.1".to_string(),
                    port: 8080,
                    ready: true,
                    node_name: None,
                    pod_name: None,
                },
                ServiceEndpoint {
                    ip: "10.0.0.2".to_string(),
                    port: 8080,
                    ready: false,
                    node_name: None,
                    pod_name: None,
                },
            ],
        };

        let healthy = K8sServiceDiscovery::get_healthy_endpoints(&service);
        assert_eq!(healthy.len(), 1);
        assert_eq!(healthy[0].ip, "10.0.0.1");
    }
}